package com.feriki.msync.process;

import com.feriki.msync.process.sink.SinkHandler;
import com.feriki.msync.process.source.SourceHandler;
import com.feriki.msync.constants.ConnectType;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.concurrent.TimeUnit;

/**
 * @author luozhida
 * @create 2024-07-09
 */
@Slf4j
public class Processor implements Runnable{

    private SourceHandler sourceHandler;
    private SinkHandler sinkHandler;

    public Processor(SourceHandler sourceHandler, SinkHandler sinkHandler) {
        this.sourceHandler = sourceHandler;
        this.sinkHandler = sinkHandler;
    }

    @Override
    public void run() {
        long startTimeMs = System.currentTimeMillis();
        Pair<Boolean, Long> result = sourceHandler.readAndHandle(sinkHandler);
        if (result.getLeft() && result.getRight() != null && result.getRight() > 0) {
            long usedMill = System.currentTimeMillis() - startTimeMs;
            log.info("成功从源【{}】同步到目标【{}】共{}条记录,耗时{}毫秒,速率约{}条/秒", sourceHandler.getName(),
                    sinkHandler.getName(), result.getRight(), usedMill,
                    (new BigDecimal(result.getRight()))
                            .divide(new BigDecimal(usedMill<=1000 ? 1 : TimeUnit.MILLISECONDS.toSeconds(usedMill)), 2, RoundingMode.HALF_DOWN));

            if (ConnectType.MILVUS.equalsIgnoreCase(sinkHandler.getType())) {
                log.info("由于同步工具无法获取到索引信息,请在同步完成后手工为【{}】创建索引并load进内存", sinkHandler.getName());
            }
        }
    }
}
